package mqtt

import akka.actor.typed.ActorSystem
import akka.actor.typed.scaladsl.Behaviors
import akka.stream.alpakka.mqtt.scaladsl.MqttSink
import akka.stream.alpakka.mqtt.{MqttConnectionSettings, MqttMessage, MqttQoS}
import akka.stream.scaladsl.{Sink, Source}
import akka.stream.{Materializer, SystemMaterializer}
import akka.util.ByteString
import akka.{Done, NotUsed}
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import spray.json._
import mqtt.MqttUtils._
import EdgeDomain._

import java.sql.Timestamp
import java.util.{Date, UUID}
import scala.concurrent.Future
import scala.concurrent.duration.DurationInt
import scala.language.postfixOps

object MqttClientDemo extends EdgeDomainJsonProtocol {

  implicit val system: ActorSystem[NotUsed] = ActorSystem(
    Behaviors.empty[NotUsed], "MqttClient"
  )
  implicit val mat: Materializer = SystemMaterializer(system).materializer

  // huawei IoTDA settings
  val serverIp = "121.36.42.100"

  val deviceTypes: List[String] = List("mq2", "dht11", "bmp280")
  deviceTypes.flatMap { deviceType: String =>
    val credentials: List[Credential] = readCredentials(deviceType)
    credentials.map { credential: Credential =>
      val deviceId: String = credential.device_id
      val deviceDomain: String = credential.device_domain
      val timestamp = new Timestamp(new Date().getTime)
      val encryptedSecret: String = HmacSHA256Encryption(credential.secret, timestamp)

      val connectionSettings: MqttConnectionSettings = MqttConnectionSettings(
        broker = s"tcp://$serverIp:1883",
        clientId = Vector(
          deviceId, // device ID
          "0",      // device identifier type
          "1",      // password signature type
          MqttUtils.getAuthTimestamp(timestamp) // timestamp
        ).mkString("_"),
        persistence = new MemoryPersistence
      )
        .withCleanSession(false)
        .withAuth(deviceId, encryptedSecret)
        .withKeepAliveInterval(100 seconds)
        .withConnectionTimeout(5000 seconds)
        .withAutomaticReconnect(true)

      val userId: String = UUID.randomUUID().toString
      val simpleDHT11Report: DHT11 = DHT11(deviceId, userId, 32.0, 17.0, status = true)
      val simpleMessage: String = Map(
        "services" -> List(DHT11Report(
          properties = simpleDHT11Report,
          event_time = MqttUtils.getEventTimestamp(timestamp)
        ))
      ).toJson.prettyPrint

      val topic: String = "$oc/devices/" + s"$deviceId/sys/properties/report"
      val payload: MqttMessage = MqttMessage(
        topic,
        ByteString(simpleMessage)
      )

      val sink: Sink[MqttMessage, Future[Done]] =
        MqttSink(connectionSettings, MqttQoS.AtLeastOnce)

      Source
        .single(payload)
        .to(sink)
        .run()
    }
  }
}
